package com.iotmonitor.iot.emq;

import com.iotmonitor.iot.config.EmqConfig;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Lazy;
import org.springframework.stereotype.Component;

import java.util.UUID;

/**
 * EMQ客户端类
 */
@Component
public class EmqClient {
    private static final Logger log = LoggerFactory.getLogger(EmqClient.class);

    @Autowired
    private EmqConfig emqConfig;

    // 客户端连接
    private MqttClient mqttClient;

    @Autowired
    @Lazy
    private EmqMsgProcess emqMsgProcess;

    /**
     * 连接emq
     */
    public void connect() {
        try {
            mqttClient = new MqttClient(emqConfig.getMqttServerUrl(), "monitor." + UUID.randomUUID());
            mqttClient.setCallback(emqMsgProcess);
            mqttClient.connect();

        } catch (MqttException e) {
            e.printStackTrace();
        }
    }


    /**
     * 发送消息
     *
     * @param topic 主题
     * @param msg   消息
     */
    public void publish(String topic, String msg) {
        MqttMessage mqttMessage = new MqttMessage(msg.getBytes());
        try {
            mqttClient.getTopic(topic).publish(mqttMessage);
        } catch (MqttException e) {
            e.printStackTrace();
            log.error("发送消息异常");
        }
    }


    /**
     * 主题订阅
     *
     * @param topicName
     * @throws MqttException
     */
    public void subscribe(String topicName) throws MqttException {
        mqttClient.subscribe(topicName);
    }

    public synchronized void publishSynchronized(String topic, String message) {
        this.publish(topic, message);
    }


}
